package com.bjtu.boatmq.client.consumer;

import cn.boat.router.consumer.*;
import org.apache.zookeeper.ZooKeeper;

/**
 *
 * @author 
 * @create 2020-06-11 22:51
 */
public class Consumer2Router {
    private String ip;
    private String port;
    private String address; // "ip:port"
    private String topicName;// 创建Consumer时指定的topic
    private ZooKeeper zooKeeperConnection = null;
    private RegisterCenterImpl registerCenter = null;

    private final String consumerGroupId="consumerGroup001";
    private final String consumerId="consumer1";

    private ConsumerConnectZookeeper connectZookeeper;
    private SubscribeTopcisImpl subscribeTopcis;

    public Consumer2Router(String topicName){
        this.topicName = topicName;
        connectZookeeper = new ConsumerConnectZookeeper();
        subscribeTopcis = new SubscribeTopcisImpl();
    }

    /**
     * 与Router建立连接
     * @throws Exception
     */
    public void connectZookeeper() throws Exception {
        zooKeeperConnection = connectZookeeper.connect();
    }

    /**
     * 注册节点
     * @throws Exception
     */
    public void register() throws Exception {
        registerCenter = new RegisterCenterImpl(zooKeeperConnection);
        registerCenter.register(consumerGroupId,consumerId);
    }

    /**
     * 订阅topic
     * @throws Exception
     */
    public ConsumerMetaData subscribe() throws Exception {
       return subscribeTopcis.subscribeTopics(zooKeeperConnection,consumerGroupId,consumerId,topicName,1);
    }


    public String getIp() {
        return ip;
    }

    public void setIp(String ip) {
        this.ip = ip;
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String ip,String port) {
        this.address = ip + ":" + port;
    }

    public String getPort() {
        return port;
    }

    public void setPort(String port) {
        this.port = port;
    }
}
